package operator;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;

/**
 * FlatMap 算子
 */
public class FlatMapDemo2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 1,2,3,4,5,6,7
        DataStreamSource<String> data = env.socketTextStream("192.168.56.10", 8888);

        SingleOutputStreamOperator<Integer> nums = data.transform("MyFlatMap", Types.INT, new MyStreamFlatMap());

        nums.print();

        env.execute("FlatMap Demo");
    }

    // 自定义StreamFlatMap
    public static class MyStreamFlatMap extends AbstractStreamOperator<Integer> implements OneInputStreamOperator<String, Integer> {


        @Override
        public void processElement(StreamRecord<String> element) throws Exception {
            java.lang.String[] words = element.getValue().split(",");
            for (String word : words) {
                int i = Integer.parseInt(word);
                if (i % 2 == 0) {
                    output.collect(element.replace(i));
                }
            }
        }
    }
}
